package com.chb.storm.gpstrident;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.hadoop.hbase.client.Durability;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import storm.trident.operation.TridentCollector;
import storm.trident.state.State;
import storm.trident.tuple.TridentTuple;
import backtype.storm.topology.FailedException;

import com.alibaba.fastjson.JSON;
import com.common.utils.AdDateFormatUtils;
import com.common.utils.MobileChbLogConstant;
import com.common.utils.TableUtils;
import com.mobilechb.bean.DriveAllBean;
import com.mobilechb.bean.GpsInfoBean;
import com.mobilechb.bean.IdentificationBean;

/**
 * create 'storm_ns:tridentgps', { NAME => 'd', COMPRESSION => 'snappy'}
 * @author Administrator
 *
 */
public class GpsHbaseState implements State {

	private static final Logger LOG = LoggerFactory.getLogger(GpsHbaseState.class);
	
	private Map map;
	
	private int numPartitions;
	
	private int partitionIndex;
	
	private static final byte[] FAMILY_C = Bytes.toBytes("d");
	
	private static final byte[] LNG_Q = Bytes.toBytes("lng");
	
	private static final byte[] LAT_Q = Bytes.toBytes("lat");
	
	private static final byte[] T_Q = Bytes.toBytes("t");
	
	private static final byte[] S_Q = Bytes.toBytes("s");
	
	private static final byte[] TYPE_Q = Bytes.toBytes("type");
	
	public GpsHbaseState(Map map, int numPartitions, int partitionIndex) {
		super();
		this.map = map;
		this.numPartitions = numPartitions;
		this.partitionIndex = partitionIndex;
	}

	@Override
	public void beginCommit(Long txid) {
		LOG.debug("beginCommit txid===={}", txid);
	}

	@Override
	public void commit(Long txid) {
		LOG.debug("commit txid===={}", txid);
	}

	public void saveGps(List<TridentTuple> tuples, TridentCollector collector)
	{
		for (TridentTuple input : tuples)
		{
			String word = (String) input.getValueByField("word");
			if (StringUtils.isBlank(word))
				continue;
			
			long currentTs = System.currentTimeMillis()/1000;
			ArrayList<Put> gpsList = new ArrayList<Put>();
			Put gpsPut = new Put(Bytes.toBytes(word), currentTs);
			gpsPut.setDurability(Durability.SYNC_WAL);
			gpsPut.addColumn(FAMILY_C, LNG_Q, Bytes.toBytes(System.currentTimeMillis()));
			gpsList.add(gpsPut);
			
			try(
				Table table = TableUtils.getTable("storm_ns:tridentgps");
			)
			{
				Object[] results = new Object[1];
				table.batch(gpsList, results);
			} catch (IOException | InterruptedException e) {
				collector.reportError(e);
				throw new FailedException(e);
			}
		}
	}
	
	/**
	 * 如果userID没有11位，则补齐到11位，并返回
	 * @param userId
	 * @return
	 */
	private static String insureUserId(String userId)
	{
		while (StringUtils.length(userId) < 11)
		{
			userId = "0" + userId;
		}
		return userId;
	}
}
